package com.pucong.gitee.database.backup.business.service.util.fileService.pool;

import lombok.extern.slf4j.Slf4j;

import java.util.HashMap;
import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.LinkedBlockingQueue;

/**
 * 自定义ftp连接池
 * @author PC
 */
@Slf4j
public class FtpConnectionManager extends Thread {

    // 1分钟检测一次连接
    private static final long CONN_CHECK_TIME = 1 * 60 * 1000L;

    //连接池
    private static ConcurrentHashMap<String, LinkedBlockingQueue<FtpClientWapper>> FTP_POOL = new ConcurrentHashMap<>();

    //一个连接最大线程数
    private static Map<String, Integer> CONNECTION_NUMLIMIT = new HashMap<>();

    //最大连接数
    private static final Integer MAX_CONNECTION_NUM = 10;

    /**
     * 定时任务检测连接池里面长时间未使用的连接
     */
    @Override
    public void run() {
        while (true) {
            try {
                Thread.sleep(CONN_CHECK_TIME);
            } catch (InterruptedException e) {
                Thread.currentThread().interrupt();
                log.error("sleep error", e);
            }
            int size = 0;
            for (String key : FTP_POOL.keySet()) {
                LinkedBlockingQueue<FtpClientWapper> queue = FTP_POOL.get(key);
                log.info("---------------------开始校验连接池中是否有过期连接，size:【{}】 key:【{}】 ---------------------", queue.size(), key);
                synchronized (FtpConnectionManager.class) {
                    Integer num = 0;
                    // 头部的肯定是时间最早的，只需要判断头部的时间是否超时即可
                    while (queue.peek() != null && (queue.peek().checkTimeOut() || !queue.peek().validateObject())) {
                        FtpClientWapper ftpClientWapper = queue.poll();
                        try {
                            log.info("关闭空闲jdbc连接:{}", key);
                            ftpClientWapper.destroyObject();
                        } catch (Exception e) {
                            log.error("关闭连接失败", e);
                        }
                        num--;
                    }
                    CONNECTION_NUMLIMIT.put(key, CONNECTION_NUMLIMIT.get(key) + num);
                }
                size += queue.size();
            }
            log.info("----------------------本次校验完毕，连接池总连接数量为【{}】----------------------", size);
        }
    }

    private static String getKey(String sessionId, String ip, int port, String username, String password) {
        return sessionId + "-" + ip + ":" + port + "-" + username + "-" + password;
    }

    /**
     * 获取连接
     *
     * @param sessionId
     * @param ip
     * @param port
     * @param username
     * @param password
     * @return
     */
    public static FtpClientWapper getConnection(String sessionId, String ip, int port, String username, String password) {
        try {
            FtpClientWapper ftpClientWapper = null;
            String key = getKey(sessionId, ip, port, username, password);
            // 从总的连接池获取
            LinkedBlockingQueue<FtpClientWapper> queue = FTP_POOL.get(key);
            synchronized (FtpConnectionManager.class) {
                if (queue == null) {
                    log.info("初始化key为【{}】的连接队列", key);
                    queue = new LinkedBlockingQueue<>();
                    FTP_POOL.put(key, queue);
                    CONNECTION_NUMLIMIT.put(key, 0);
                }
                Integer limitNum = CONNECTION_NUMLIMIT.get(key);
                log.info("key为【{}】对应的limitNum = 【{}】", key, limitNum);
                if (!queue.isEmpty()) {
                    log.info("key为【{}】的连接池中有可用的连接，直接获取", key);
                    return queue.take();
                }
                if (limitNum >= MAX_CONNECTION_NUM) {
                    log.info("key为【{}】超过了最大连接数，进入等待队列", key);
                    ftpClientWapper = FTP_POOL.get(key).take();
                } else {
                    log.info("key为【{}】无可用的连接，开始创建连接", key);
                    ftpClientWapper = createConnection(sessionId, ip, port, username, password);
                    CONNECTION_NUMLIMIT.put(key, CONNECTION_NUMLIMIT.get(key) + 1);
                }
            }
            return ftpClientWapper;
        } catch (Exception e) {
            log.error("getConnection error", e);
        }
        return null;
    }

    private static FtpClientWapper createConnection(String sessionId, String ip, int port, String username, String password) {
        String key = getKey(sessionId, ip, port, username, password);
        return new FtpClientWapper(FTP_POOL.get(key), ip, port, username, password);
    }
}
